/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.matrix;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.results.ResultPayload;
import cgl.granules.scheduler.StreamingService;
import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventProducer;

public class MatrixSharedMemory {

	private ConcurrentHashMap<String, int[]> entries;
	private ConcurrentHashMap<String, AtomicInteger> counts;
	private Object syncObject;
	private ResultPayload resultPayload;
	private static final MatrixSharedMemory instance = new MatrixSharedMemory();
	private StreamingService streamingService;

	private EventProducer producer;
	private String workUnitStream = "FinishedWorkUnit/MatrixElement";
	private final int maxMapsPerMachine = 2000;
	private final int numOfRowsAtAtime = 1000;

	private int rowNumber = 0;

	private int numOfEntries;
	private AtomicInteger processingCounter;
	
	private MatrixSharedMemory() {
		entries = new ConcurrentHashMap<String, int[]>();
		counts = new ConcurrentHashMap<String, AtomicInteger>();
		resultPayload = ResultPayload.getInstance();
		streamingService = StreamingService.getInstance();
		numOfEntries = numOfRowsAtAtime;
		processingCounter = new AtomicInteger(0);
	}

	public static MatrixSharedMemory getInstance() {
		return instance;
	}

	public synchronized int[] getRow(String rowInfo, NBEvent nbEvent)
			throws MarshallingException {
		//System.out.println(processingCounter.incrementAndGet());
		/*
		 * Integer count = counts.get(rowInfo); if (count == null) { count = 0;
		 * } count++; counts.put(rowInfo, count);
		 */
		if (entries.containsKey(rowInfo)) {
			return entries.get(rowInfo);
		}

		byte[] payload = nbEvent.getContentPayload();
		int[] row = resultPayload.getIntArray(payload);
		if (row.length == 0) {
			System.out.println("Problem with created row. Length == 0");
		}
		entries.put(rowInfo, row);
		counts.put(rowInfo, new AtomicInteger(maxMapsPerMachine));
		return row;
	}

	public synchronized void returnRow(String rowInfo) {
		AtomicInteger count = counts.get(rowInfo);
		int value = count.decrementAndGet();
		// = count.intValue();

		if (value == 0) {
			numOfEntries--;
			counts.remove(rowInfo);
			rowNumber++;
			int[] discardRow = entries.remove(rowInfo);
			discardRow = null;

			System.out.println("Pending rows = " + numOfEntries + ", processed row ["
					+ rowNumber + "] Mem=" + getMemoryUtilization() + " MB");

			if (numOfEntries == 0) {

				try {
					if (!entries.isEmpty()) {
						System.out
								.println("Entries not empty =" + entries.size());
					}
					publishToStream(workUnitStream);
				} catch (Exception e) {
					e.printStackTrace();
				}
				numOfEntries = numOfRowsAtAtime;
				//rowNumber = 0;
				System.gc();
			}

			return;
		}// end if count == 0

		//counts.put(rowInfo, count);
	}


	private long getMemoryUtilization() {
		Runtime runtime = Runtime.getRuntime();
		long totalMemory = runtime.totalMemory();
		long freeMemory = runtime.freeMemory();

		long memUtilization = totalMemory - freeMemory;
		return (memUtilization / (1024 * 1024));
	}

	private void publishToStream(String publishStream)
			throws CommunicationsException, ServiceException,
			MarshallingException {
		if (producer == null) {
			producer = streamingService.createProducer();
		}

		NBEvent nbEvent;

		nbEvent = producer.generateEvent(
				TemplateProfileAndSynopsisTypes.STRING, publishStream, "done"
						.getBytes());
		producer.publishEvent(nbEvent);

	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

}
